package org.ws.task.item;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.ws.task.listener.TaskItemListenerManager;
import org.ws.task.model.protocol.CommitRequest;
import org.ws.task.model.protocol.CommitResponse;
import org.ws.task.model.protocol.PullItemRequest;
import org.ws.task.model.protocol.PullItemResponse;
import org.ws.task.model.protocol.PushItemRequest;
import org.ws.task.model.protocol.PushItemResponse;
import org.ws.task.netty.NettyClient;
import org.ws.task.model.protocol.base.RequestAction;
import org.ws.task.model.protocol.base.RequestMessage;
import org.ws.task.builder.RequestMessageBuilder;
import org.ws.task.model.protocol.base.ResponseMessage;

import java.util.concurrent.ExecutorService;
import java.util.function.Consumer;

/**
 * Worker远程链接API
 */
public class RemoteAccessApi {

    private static Logger log = LoggerFactory.getLogger(RemoteAccessApi.class.getName());

    private final NettyClient nettyClient;

    private TaskItemListenerManager listenerManager;

    public void setListenerManager(TaskItemListenerManager listenerManager) {
        this.listenerManager = listenerManager;
    }

    private final ExecutorService callbackExecutor;

    public RemoteAccessApi(NettyClient nettyClient,ExecutorService callbackExecutor) {
        this.nettyClient = nettyClient;
        this.listenerManager = new TaskItemListenerManager();
        this.callbackExecutor = callbackExecutor;
    }


    /**
     * 异步拉取任务
     * @param request
     * @param timeout
     * @param callback
     */
    public <T> void fetchItemAsync(PullItemRequest request, long timeout, Consumer<PullItemResponse<T>> callback){
        RequestMessage<PullItemRequest> requestMessage = RequestMessageBuilder.builder()
                .action(RequestAction.PULL_TASK)
                .body(request)
                .build();
        this.listenerManager.beforeFetch(request);
        this.callbackExecutor.submit(()->{
            try {
                RemoteAccessApi.this.nettyClient.async(null,requestMessage,timeout,responseFuture -> {
                    if(responseFuture.isSendSuccess()){
                        ResponseMessage<PullItemResponse<T>> response = responseFuture.getResponse();
                        if(response.isSuccess()){
                            final PullItemResponse<T> pullItemResponse = response.getBodyObject();
                            callback.accept(pullItemResponse);
                            RemoteAccessApi.this.listenerManager.afterFetch(request, pullItemResponse);
                        }
                        else {
                            log.warn("taskName:{} request:{} response failed.",request.getTaskName(),requestMessage.getId());
                            RemoteAccessApi.this.listenerManager.onFetchError(request,null);
                            callback.accept(null);
                        }
                    }
                    else {
                        log.warn("taskName:{} request:{} send response failed.",request.getTaskName(),requestMessage.getId());
                        RemoteAccessApi.this.listenerManager.onFetchError(request,responseFuture.getCause());
                        callback.accept(null);

                    }
                });
            } catch (Exception e) {
                log.error(e.getMessage(),e);
                RemoteAccessApi.this.listenerManager.onFetchError(request,e);
                callback.accept(null);
            }
        });
    }

    /**
     * 同步拉取任务
     * @param request
     * @param timeout
     * @param <T>
     * @return
     */
    public <T> PullItemResponse<T> fetchItemSync(PullItemRequest request, long timeout){
        RequestMessage<PullItemRequest> requestMessage = RequestMessageBuilder.builder()
                .action(RequestAction.PULL_TASK)
                .body(request)
                .build();
        this.listenerManager.beforeFetch(request);
        try {
            ResponseMessage<PullItemResponse<T>> responseMessage = this.nettyClient.sync(null, requestMessage, timeout);
            if(responseMessage.isSuccess() && responseMessage.getBodyObject()!= null){
                final PullItemResponse<T> pullItemResponse = responseMessage.getBodyObject();
                this.listenerManager.afterFetch(request, pullItemResponse);
                return pullItemResponse;
            }
            else {
                log.warn("taskName:{} request:{} response failed.",request.getTaskName(),requestMessage.getId());
                this.listenerManager.onFetchError(request,null);
            }
        } catch (Exception e) {
            log.error(e.getMessage(),e);
            RemoteAccessApi.this.listenerManager.onFetchError(request,e);
        }
        return null;
    }

    /**
     * 异步提交状态
     * @param request
     * @param timeout
     * @param callback
     */
    public void commitStatusAsync(CommitRequest request, long timeout, Consumer<CommitResponse> callback){
        RequestMessage<PullItemRequest> requestMessage = RequestMessageBuilder.builder()
                .action(RequestAction.COMMIT_STATUS)
                .body(request)
                .build();
        this.listenerManager.beforeCommit(request);
        this.callbackExecutor.submit(()->{
            try {
                RemoteAccessApi.this.nettyClient.async(null,requestMessage,timeout,responseFuture -> {
                    if(responseFuture.isSendSuccess()){
                        ResponseMessage<CommitResponse> response = responseFuture.getResponse();
                        if(response.isSuccess()){
                            final CommitResponse commitResponse = response.getBodyObject();
                            log.debug("commitResponse:{}",commitResponse);
                            callback.accept(commitResponse);
                            RemoteAccessApi.this.listenerManager.afterCommit(request,commitResponse);
                        }
                        else {
                            log.warn("taskName:{} commit:{} response failed.",request.getTaskName(),requestMessage.getId());
                            RemoteAccessApi.this.listenerManager.onCommonError(request,null);
                            callback.accept(null);
                        }
                    }
                    else {
                        log.warn("taskName:{} commit:{} send response failed.",request.getTaskName(),requestMessage.getId());
                        RemoteAccessApi.this.listenerManager.onCommonError(request,responseFuture.getCause());
                        callback.accept(null);

                    }
                });
            } catch (Exception e) {
                log.error(e.getMessage(),e);
                RemoteAccessApi.this.listenerManager.onCommonError(request,e);
                callback.accept(null);
            }
        });
    }

    /**
     * 同步提交状态
     * @param request
     * @param timeout
     * @return
     */
    public CommitResponse commitStatusSync(CommitRequest request,long timeout){
        RequestMessage<PullItemRequest> requestMessage = RequestMessageBuilder.builder()
                .action(RequestAction.COMMIT_STATUS)
                .body(request)
                .build();
        this.listenerManager.beforeCommit(request);
        try {
            ResponseMessage<CommitResponse> responseMessage = this.nettyClient.sync(null, requestMessage, timeout);
            if(responseMessage.isSuccess()){
                final CommitResponse commitResponse = responseMessage.getBodyObject();
                log.debug("commitResponse:{}",commitResponse);
                this.listenerManager.afterCommit(request,commitResponse);
                return commitResponse;
            }
            else {
                log.warn("taskName:{} request:{} response failed.",request.getTaskName(),requestMessage.getId());
                this.listenerManager.onCommonError(request,null);
            }
        } catch (Exception e) {
            log.error(e.getMessage(),e);
            RemoteAccessApi.this.listenerManager.onCommonError(request,e);
        }
        return null;
    }

    /**
     * worker推送任务到leader
     * @param request
     * @param timeout
     * @return
     */
    public PushItemResponse pushItemSync(PushItemRequest request, long timeout){
        RequestMessage<PushItemRequest> requestMessage = RequestMessageBuilder.builder()
                .action(RequestAction.PUSH_TASK)
                .body(request)
                .build();
        this.listenerManager.beforePush(request);
        try {
            ResponseMessage<PushItemResponse> responseMessage = this.nettyClient.sync(null, requestMessage, timeout);
            if(responseMessage != null && responseMessage.isSuccess()){
                final PushItemResponse pushItemResponse = responseMessage.getBodyObject();
                log.debug("pushItemResponse:{}",pushItemResponse);
                this.listenerManager.afterPush(request,pushItemResponse);
                return pushItemResponse;
            }
            else {
                log.warn("taskName:{} request:{} response failed.",request.getTaskName(),requestMessage.getId());
                this.listenerManager.onPushError(request,null);
            }
        } catch (Exception e) {
            log.error(e.getMessage(),e);
            RemoteAccessApi.this.listenerManager.onPushError(request,e);
        }
        return null;
    }

    /**
     *
     */
    public void pushItemAsync(PushItemRequest request,long timeout,Consumer<PushItemResponse> callback){
        RequestMessage<PushItemRequest> requestMessage = RequestMessageBuilder.builder()
                .action(RequestAction.PUSH_TASK)
                .body(request)
                .build();
        this.listenerManager.beforePush(request);
        this.callbackExecutor.submit(()->{
            try {
                RemoteAccessApi.this.nettyClient.async(null,requestMessage,timeout,responseFuture -> {
                    if(responseFuture.isSendSuccess()){
                        ResponseMessage<PushItemResponse> response = responseFuture.getResponse();
                        if(response.isSuccess()){
                            final PushItemResponse pushItemResponse = response.getBodyObject();
                            log.debug("pushItemResponse:{}",pushItemResponse);
                            callback.accept(pushItemResponse);
                            RemoteAccessApi.this.listenerManager.afterPush(request,pushItemResponse);
                        }
                        else {
                            log.warn("taskName:{} push:{} response failed.",request.getTaskName(),requestMessage.getId());
                            RemoteAccessApi.this.listenerManager.onPushError(request,null);
                            callback.accept(null);
                        }
                    }
                    else {
                        log.warn("taskName:{} push:{} send response failed.",request.getTaskName(),requestMessage.getId());
                        RemoteAccessApi.this.listenerManager.onPushError(request,responseFuture.getCause());
                        callback.accept(null);

                    }
                });
            } catch (Exception e) {
                log.error(e.getMessage(),e);
                RemoteAccessApi.this.listenerManager.onPushError(request,e);
                callback.accept(null);
            }
        });
    }
}
